package com.example.base;

import com.example.entity.Customer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.formats.json.JsonDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author wangJinLong
 * Date 2025/8/22 14:34
 **/
public class FlinkStreamOldJdbc {
    public static void main(String[] args) {

    }

    private static void kafkaToJdbc() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KafkaSource<Customer> source = KafkaSource.<Customer>builder()
                .setBootstrapServers("106.54.174.109:19092")
                .setTopics("data-user")
                .setGroupId("my-group1")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new JsonDeserializationSchema<>(Customer.class))
                .build();

        DataStreamSource<Customer> dataStreamSource =
                env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        dataStreamSource.print();

        dataStreamSource.addSink(
                JdbcSink.sink(
                        "insert into customers_kafka (featuresId, siteSn, captureTime) values (?, ?, ?)",
                        (JdbcStatementBuilder<Customer>) (preparedStatement, customer) -> {
                            preparedStatement.setString(1, customer.getFeaturesId());
                            preparedStatement.setString(2, customer.getSiteSn());
                            preparedStatement.setString(3, Long.toString(customer.getCaptureTime()));
                        },
                        JdbcExecutionOptions.builder()
                                .withMaxRetries(1)
                                .withBatchSize(10)
                                .withBatchIntervalMs(5000)
                                .build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withDriverName("com.mysql.jdbc.Driver")
                                .withUrl("jdbc:mysql://106.54.174.109:13306/my_cdc")
                                .withUsername("root")
                                .withPassword("mysql@123456")
                                .build()
                )
        );

        env.execute();
    }
}
